package com.example.transformation;

import com.example.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Created with IntelliJ IDEA.
 * ClassName: MapDemo
 * Package: com.example.transformation
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-18
 * Time: 16:03
 */

//过滤
public class FilterDemo {
    public static void main(String[] args) throws Exception {

        //转换算子 map 映射
        //需求 提取WaterSensor中的id字段的功能

        //1.创建环境
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //fromElements 是直接赋值数据
        DataStreamSource<WaterSensor> data = env.fromElements(
                //赋值WaterSensor对象的数据
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );


        //传感器id为sensor_1的数据过滤出来
        final SingleOutputStreamOperator<WaterSensor> filter = data.filter(new FilterFunction<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor waterSensor) throws Exception {
                if ("s1".equals(waterSensor.getId())) {
                    return true;
                } else {
                    return false;
                }
            }
        });

        filter.print();
        env.execute();

    }
}
